package com.shujia.flink.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object Demo2MysqlSInk {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentDS: DataStream[String] = env.readTextFile("flink/data/students.txt")

    studentDS.addSink(new MysqlSink)

    env.execute()

  }

}

/**
  * RichSinkFunction： 多了open和close方法
  * SinkFunction
  *
  */

class MysqlSink extends RichSinkFunction[String] {
  /**
    * 在open中创建链接，
    *
    */
  var con: Connection = _

  override def open(parameters: Configuration): Unit = {
    println("创建mysql链接")
    Class.forName("com.mysql.jdbc.Driver")
    con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=utf-8", "root", "123456")
  }

  override def close(): Unit = {
    //关闭链接
    con.close()
  }

  override def invoke(line: String, context: SinkFunction.Context[_]): Unit = {
    val split: Array[String] = line.split(",")
    val id: String = split(0)
    val name: String = split(1)
    val age: Int = split(2).toInt
    val gender: String = split(3)
    val clazz: String = split(4)


    //构建sql
    val stat: PreparedStatement = con.prepareStatement("insert into  student(id,name,age,gender,clazz) values(?,?,?,?,?)")

    //设置参数
    stat.setString(1, id)
    stat.setString(2, name)
    stat.setInt(3, age)
    stat.setString(4, gender)
    stat.setString(5, clazz)

    //执行插入
    stat.execute()

  }
}
